package com.yifeng.repo.controller.rocketmq.message.consume;

import com.talkyun.utils.Looper;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;
import java.util.function.Consumer;

/**
 * Created by daibing on 2023/8/6.
 */
@Slf4j
public class SubscribePullConsumerLooper extends Looper {
    private final String topic;
    private final DefaultLitePullConsumer pullConsumer;
    private final Consumer<List<MessageExt>> msgListConsumer;

    public SubscribePullConsumerLooper(String name, int interval, int errorInterval, String topic,
                                       DefaultLitePullConsumer pullConsumer, Consumer<List<MessageExt>> msgListConsumer) {
        super(name, interval, errorInterval);
        this.topic = topic;
        this.pullConsumer = pullConsumer;
        this.msgListConsumer = msgListConsumer;
    }

    @Override
    protected void loop() throws Throwable {
        // 拉取&消费消息
        List<MessageExt> messages = pullConsumer.poll();
        msgListConsumer.accept(messages);
        messages.clear();
    }

    @Override
    protected void loopThrowable(Throwable t) {
        log.warn("consumer {} failed: error=", topic, t);
    }

    @Override
    protected void clear() {
        pullConsumer.unsubscribe(topic);
        pullConsumer.shutdown();
    }

    public DefaultLitePullConsumer getPullConsumer() {
        return pullConsumer;
    }

}
